package com.zyh.kafka.day03;

import com.zyh.kafka.util.KafkaConsumerUtils;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;

public class KafkaConsumerG3 {
    public static void main(String[] args) {
        Properties consumerConfigs = new Properties();
        consumerConfigs.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"kafka24:9092");
        consumerConfigs.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerConfigs.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerConfigs.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"g5");
        // 设置一个新的消费组第一次消费topic时，从队头开始消费
        //consumerConfigs.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        // 禁用自动提交
        //consumerConfigs.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(consumerConfigs);
        // 订阅消息
        kafkaConsumer.subscribe(Arrays.asList("topic-a"));
        // 动态获取订阅的topic的所有分区,需要先和kafka客户端建立链接
        Set<TopicPartition> assignment = kafkaConsumer.assignment();
        while(assignment.isEmpty()){
            kafkaConsumer.poll(100);
            assignment = kafkaConsumer.assignment();
        }
        // 设置每个分区的起始消费位移位置
        for (TopicPartition topicPartition : assignment) {
            kafkaConsumer.seek(topicPartition,10);
        }
        try {
            while (true){
                // 拉取消息
                ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);
                // 处理消息
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("record = " + record);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            kafkaConsumer.close();
        }
    }
}
